package bootstrap

import (
	"app/pkg/queue"
	"app/pkg/queue/drivers"
	"context"
	"github.com/go-redis/redis/v8"
	"log"
	"sync"
)

var l sync.Mutex
var queues map[string]queue.IQueue

func init() {
	queues = make(map[string]queue.IQueue)
}

func RegisterQueue(q queue.IQueue) {
	l.Lock()
	defer l.Unlock()
	topic := q.Topic()
	if _, ok := queues[topic]; ok {
		return
	}
	queues[topic] = q
}

// RegisterQueueDrivers 注册队列驱动
func RegisterQueueDrivers() {
	//redis驱动
	queue.RegisterDriver("redis", drivers.NewRedis(&redis.Options{}))
	queue.RegisterDriver("redis-stream", drivers.NewRedisStream(&redis.Options{}))
	//文件系统驱动
	queue.RegisterDriver("disk", drivers.NewDisk("/app/storage/queue"))
	//TODO 注册其他队列驱动
}

// StartQueue 启动所有已注册的消费者监听
func QueueListen(ctx context.Context) {
	for _, q := range queues {
		// TODO 可以增加判断队列是否启用，如果没启用就不去消费
		go func(q queue.IQueue) {
			var topic = q.Topic()
			var driver = queue.GetDriver(q.Driver())
			err := driver.Subscribe(ctx, topic, q.Options(), func(ctx context.Context, channel <-chan queue.QueueMessgae) {
				for {
					select {
					case msg := <-channel:
						//消费成功
						if ok := q.Consume(ctx, msg); ok {
							driver.Ack(ctx, q, msg)
							continue
						}
						//消费失败
						if q.Retry() > 0 && msg.GetHeader(queue.HeaderDelivery).(int) < q.Retry() {
							//重试
							driver.Nack(ctx, q, msg)
						} else {
							//进入死信队列
							driver.Reject(ctx, q, msg)
						}
					case <-ctx.Done():
						// 队列结束
					}
				}
			})
			if err != nil {
				log.Printf("队列【%s】监听错误:%s", topic, err)
			}
		}(q)
	}
}
